package org.cybzacg.encryption.batch.impl;

import org.cybzacg.encryption.batch.BatchProcessor;
import org.cybzacg.encryption.context.EncryptionContext;
import org.cybzacg.encryption.model.EncryptionResult;
import org.cybzacg.encryption.model.ValidationResult;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
import java.util.stream.Stream;

/**
 * 流式批处理器实现
 * 使用Java Stream API进行内存优化的批处理
 * 
 * @author encryption team
 * @version 1.0
 */
public class StreamBatchProcessor<T, R> implements BatchProcessor<T, R> {
    
    // 配置参数
    private volatile int batchSize = 100;
    private volatile int maxConcurrency = Runtime.getRuntime().availableProcessors();
    private volatile long timeout = 30000; // 30秒
    private volatile boolean parallelProcessing = true;
    
    // 统计信息
    private final AtomicLong totalBatches = new AtomicLong(0);
    private final AtomicLong totalProcessed = new AtomicLong(0);
    private final AtomicLong totalSuccess = new AtomicLong(0);
    private final AtomicLong totalFailures = new AtomicLong(0);
    private final AtomicLong totalProcessingTime = new AtomicLong(0);
    private volatile long maxProcessingTime = 0;
    private volatile long minProcessingTime = Long.MAX_VALUE;
    
    // 监听器
    private final List<BatchListener<T, R>> listeners = new CopyOnWriteArrayList<>();
    
    /**
     * 构造函数
     */
    public StreamBatchProcessor() {
        // 默认构造函数
    }
    
    /**
     * 构造函数
     * 
     * @param batchSize 批次大小
     */
    public StreamBatchProcessor(int batchSize) {
        this.batchSize = batchSize;
    }
    
    /**
     * 构造函数
     * 
     * @param batchSize 批次大小
     * @param parallelProcessing 是否启用并行处理
     */
    public StreamBatchProcessor(int batchSize, boolean parallelProcessing) {
        this.batchSize = batchSize;
        this.parallelProcessing = parallelProcessing;
    }
    
    @Override
    public List<R> processBatch(List<T> dataList, EncryptionContext context, Function<T, R> processor) {
        if (dataList == null || dataList.isEmpty()) {
            return Collections.emptyList();
        }
        
        long startTime = System.currentTimeMillis();
        
        // 验证配置
        ValidationResult validationResult = validateConfiguration(context);
        if (!validationResult.isValid()) {
            throw new IllegalArgumentException("Invalid configuration: " + validationResult.getMessage());
        }
        
        // 通知开始
        notifyBatchStart(dataList.size());
        
        try {
            // 使用流式处理
            Stream<T> stream = dataList.stream();
            
            if (parallelProcessing) {
                stream = stream.parallel();
            }
            
            List<R> results = stream
                    .map(data -> {
                        try {
                            return processor.apply(data);
                        } catch (Exception e) {
                            // 记录失败
                            totalFailures.incrementAndGet();
                            return null;
                        }
                    })
                    .filter(Objects::nonNull)
                    .collect(ArrayList::new, ArrayList::add, ArrayList::addAll);
            
            // 更新统计信息
            long processingTime = System.currentTimeMillis() - startTime;
            updateStatistics(processingTime, dataList.size(), results.size());
            
            // 通知完成
            notifyBatchComplete(createBatchResult(results, dataList.size(), processingTime));
            
            return results;
            
        } catch (Exception e) {
            // 通知错误
            notifyBatchError(createBatchError(e));
            throw new RuntimeException("Stream batch processing failed", e);
        }
    }
    
    @Override
    public CompletableFuture<List<R>> processBatchAsync(List<T> dataList, EncryptionContext context, Function<T, R> processor) {
        return CompletableFuture.supplyAsync(() -> 
            processBatch(dataList, context, processor)
        );
    }
    
    @Override
    public List<EncryptionResult> encryptBatch(List<byte[]> dataList, EncryptionContext context) {
        if (dataList == null || dataList.isEmpty()) {
            return Collections.emptyList();
        }
        
        long startTime = System.currentTimeMillis();
        
        // 验证配置
        ValidationResult validationResult = validateConfiguration(context);
        if (!validationResult.isValid()) {
            throw new IllegalArgumentException("Invalid configuration: " + validationResult.getMessage());
        }
        
        try {
            // 使用流式处理加密
            Stream<byte[]> stream = dataList.stream();
            
            if (parallelProcessing) {
                stream = stream.parallel();
            }
            
            List<EncryptionResult> results = stream
                    .map(data -> processSingleEncryption(data, context))
                    .filter(Objects::nonNull)
                    .collect(ArrayList::new, ArrayList::add, ArrayList::addAll);
            
            // 更新统计信息
            long processingTime = System.currentTimeMillis() - startTime;
            updateStatistics(processingTime, dataList.size(), results.size());
            
            return results;
            
        } catch (Exception e) {
            throw new RuntimeException("Stream encryption batch processing failed", e);
        }
    }
    
    @Override
    public CompletableFuture<List<EncryptionResult>> encryptBatchAsync(List<byte[]> dataList, EncryptionContext context) {
        return CompletableFuture.supplyAsync(() -> 
            encryptBatch(dataList, context)
        );
    }
    
    @Override
    public List<byte[]> decryptBatch(List<byte[]> dataList, EncryptionContext context) {
        if (dataList == null || dataList.isEmpty()) {
            return Collections.emptyList();
        }
        
        long startTime = System.currentTimeMillis();
        
        // 验证配置
        ValidationResult validationResult = validateConfiguration(context);
        if (!validationResult.isValid()) {
            throw new IllegalArgumentException("Invalid configuration: " + validationResult.getMessage());
        }
        
        try {
            // 使用流式处理解密
            Stream<byte[]> stream = dataList.stream();
            
            if (parallelProcessing) {
                stream = stream.parallel();
            }
            
            List<byte[]> results = stream
                    .map(data -> processSingleDecryption(data, context))
                    .filter(Objects::nonNull)
                    .collect(ArrayList::new, ArrayList::add, ArrayList::addAll);
            
            // 更新统计信息
            long processingTime = System.currentTimeMillis() - startTime;
            updateStatistics(processingTime, dataList.size(), results.size());
            
            return results;
            
        } catch (Exception e) {
            throw new RuntimeException("Stream decryption batch processing failed", e);
        }
    }
    
    @Override
    public CompletableFuture<List<byte[]>> decryptBatchAsync(List<byte[]> dataList, EncryptionContext context) {
        return CompletableFuture.supplyAsync(() -> 
            decryptBatch(dataList, context)
        );
    }
    
    @Override
    public int getBatchSize() {
        return batchSize;
    }
    
    @Override
    public void setBatchSize(int batchSize) {
        if (batchSize <= 0) {
            throw new IllegalArgumentException("Batch size must be positive");
        }
        this.batchSize = batchSize;
    }
    
    @Override
    public int getMaxConcurrency() {
        return maxConcurrency;
    }
    
    @Override
    public void setMaxConcurrency(int maxConcurrency) {
        if (maxConcurrency <= 0) {
            throw new IllegalArgumentException("Max concurrency must be positive");
        }
        this.maxConcurrency = maxConcurrency;
    }
    
    @Override
    public long getTimeout() {
        return timeout;
    }
    
    @Override
    public void setTimeout(long timeout) {
        if (timeout <= 0) {
            throw new IllegalArgumentException("Timeout must be positive");
        }
        this.timeout = timeout;
    }
    
    @Override
    public ValidationResult validateConfiguration(EncryptionContext context) {
        if (context == null) {
            return new ValidationResult.Builder()
                    .valid(false)
                    .message("Context cannot be null")
                    .build();
        }
        
        if (batchSize <= 0) {
            return new ValidationResult.Builder()
                    .valid(false)
                    .message("Batch size must be positive")
                    .build();
        }
        
        if (maxConcurrency <= 0) {
            return new ValidationResult.Builder()
                    .valid(false)
                    .message("Max concurrency must be positive")
                    .build();
        }
        
        if (timeout <= 0) {
            return new ValidationResult.Builder()
                    .valid(false)
                    .message("Timeout must be positive")
                    .build();
        }
        
        return new ValidationResult.Builder()
                .valid(true)
                .message("Configuration is valid")
                .build();
    }
    
    @Override
    public BatchStatistics getStatistics() {
        return new BatchStatisticsImpl(
            totalBatches.get(),
            totalProcessed.get(),
            totalSuccess.get(),
            totalFailures.get(),
            totalProcessed.get() > 0 ? (double) totalProcessingTime.get() / totalProcessed.get() : 0.0,
            maxProcessingTime,
            minProcessingTime == Long.MAX_VALUE ? 0 : minProcessingTime,
            calculateThroughput(),
            totalProcessed.get() > 0 ? (double) totalSuccess.get() / totalProcessed.get() : 0.0
        );
    }
    
    @Override
    public void resetStatistics() {
        totalBatches.set(0);
        totalProcessed.set(0);
        totalSuccess.set(0);
        totalFailures.set(0);
        totalProcessingTime.set(0);
        maxProcessingTime = 0;
        minProcessingTime = Long.MAX_VALUE;
    }
    
    /**
     * 检查是否启用并行处理
     * 
     * @return 是否启用并行处理
     */
    public boolean isParallelProcessing() {
        return parallelProcessing;
    }
    
    /**
     * 设置并行处理
     * 
     * @param parallelProcessing 是否启用并行处理
     */
    public void setParallelProcessing(boolean parallelProcessing) {
        this.parallelProcessing = parallelProcessing;
    }
    
    /**
     * 添加批处理监听器
     * 
     * @param listener 监听器
     */
    public void addListener(BatchListener<T, R> listener) {
        if (listener != null) {
            listeners.add(listener);
        }
    }
    
    /**
     * 移除批处理监听器
     * 
     * @param listener 监听器
     */
    public void removeListener(BatchListener<T, R> listener) {
        listeners.remove(listener);
    }
    
    /**
     * 处理单个加密
     */
    private EncryptionResult processSingleEncryption(byte[] data, EncryptionContext context) {
        try {
            // 这里应该调用加密引擎
            // 简化实现
            return new EncryptionResult.Builder()
                    .data(data)
                    .status(org.cybzacg.encryption.enums.EncryptionStatus.SUCCESS)
                    .algorithm(context.getAlgorithm())
                    .build();
        } catch (Exception e) {
            return new EncryptionResult.Builder()
                    .status(org.cybzacg.encryption.enums.EncryptionStatus.UNKNOWN_ERROR)
                    .errorMessage(e.getMessage())
                    .algorithm(context.getAlgorithm())
                    .build();
        }
    }
    
    /**
     * 处理单个解密
     */
    private byte[] processSingleDecryption(byte[] data, EncryptionContext context) {
        try {
            // 这里应该调用解密引擎
            // 简化实现
            return data;
        } catch (Exception e) {
            return null;
        }
    }
    
    /**
     * 更新统计信息
     */
    private void updateStatistics(long processingTime, int totalCount, int successCount) {
        totalBatches.incrementAndGet();
        totalProcessed.addAndGet(totalCount);
        totalSuccess.addAndGet(successCount);
        totalFailures.addAndGet(totalCount - successCount);
        totalProcessingTime.addAndGet(processingTime);
        
        if (processingTime > maxProcessingTime) {
            maxProcessingTime = processingTime;
        }
        
        if (processingTime < minProcessingTime) {
            minProcessingTime = processingTime;
        }
    }
    
    /**
     * 计算吞吐量
     */
    private double calculateThroughput() {
        long totalTime = totalProcessingTime.get();
        if (totalTime > 0) {
            return (double) totalProcessed.get() * 1000 / totalTime; // 操作/秒
        }
        return 0.0;
    }
    
    /**
     * 创建批处理结果
     */
    private BatchResult<R> createBatchResult(List<R> results, int totalCount, long processingTime) {
        return new BatchResultImpl<>(results, totalCount, processingTime);
    }
    
    /**
     * 创建批处理错误
     */
    private BatchError<R> createBatchError(Exception exception) {
        return new BatchErrorImpl<>(exception);
    }
    
    /**
     * 通知批处理开始
     */
    private void notifyBatchStart(int totalCount) {
        for (BatchListener<T, R> listener : listeners) {
            try {
                listener.onBatchStart(totalCount);
            } catch (Exception e) {
                // 忽略监听器异常
            }
        }
    }
    
    /**
     * 通知批处理完成
     */
    private void notifyBatchComplete(BatchResult<R> result) {
        for (BatchListener<T, R> listener : listeners) {
            try {
                listener.onBatchComplete(result);
            } catch (Exception e) {
                // 忽略监听器异常
            }
        }
    }
    
    /**
     * 通知批处理错误
     */
    private void notifyBatchError(BatchError<R> error) {
        for (BatchListener<T, R> listener : listeners) {
            try {
                listener.onBatchError(error);
            } catch (Exception e) {
                // 忽略监听器异常
            }
        }
    }
    
    /**
     * 流式批处理结果实现
     */
    private static class BatchResultImpl<R> implements BatchResult<R> {
        
        private final List<R> successResults;
        private final int totalCount;
        private final long processingTime;
        
        public BatchResultImpl(List<R> successResults, int totalCount, long processingTime) {
            this.successResults = successResults != null ? successResults : Collections.emptyList();
            this.totalCount = totalCount;
            this.processingTime = processingTime;
        }
        
        @Override
        public List<R> getSuccessResults() {
            return successResults;
        }
        
        @Override
        public List<BatchError<R>> getFailedResults() {
            return Collections.emptyList(); // 简化实现
        }
        
        @Override
        public int getTotalCount() {
            return totalCount;
        }
        
        @Override
        public int getSuccessCount() {
            return successResults.size();
        }
        
        @Override
        public int getFailureCount() {
            return totalCount - successResults.size();
        }
        
        @Override
        public double getSuccessRate() {
            return totalCount > 0 ? (double) getSuccessCount() / totalCount : 0.0;
        }
        
        @Override
        public long getProcessingTime() {
            return processingTime;
        }
        
        @Override
        public boolean isAllSuccess() {
            return getFailureCount() == 0;
        }
        
        @Override
        public boolean hasFailures() {
            return getFailureCount() > 0;
        }
    }
    
    /**
     * 流式批处理错误实现
     */
    private static class BatchErrorImpl<R> implements BatchError<R> {
        
        private final Exception exception;
        
        public BatchErrorImpl(Exception exception) {
            this.exception = exception;
        }
        
        @Override
        public Object getOriginalData() {
            return null;
        }
        
        @Override
        public String getErrorMessage() {
            return exception != null ? exception.getMessage() : "Unknown error";
        }
        
        @Override
        public Exception getException() {
            return exception;
        }
        
        @Override
        public ErrorType getErrorType() {
            return ErrorType.PROCESSING_ERROR;
        }
    }
    
    /**
     * 流式批处理统计信息实现
     */
    private static class BatchStatisticsImpl implements BatchStatistics {
        
        private final long totalBatches;
        private final long totalProcessed;
        private final long totalSuccess;
        private final long totalFailures;
        private final double averageProcessingTime;
        private final long maxProcessingTime;
        private final long minProcessingTime;
        private final double throughput;
        private final double averageSuccessRate;
        
        public BatchStatisticsImpl(long totalBatches, long totalProcessed, long totalSuccess,
                                long totalFailures, double averageProcessingTime,
                                long maxProcessingTime, long minProcessingTime,
                                double throughput, double averageSuccessRate) {
            this.totalBatches = totalBatches;
            this.totalProcessed = totalProcessed;
            this.totalSuccess = totalSuccess;
            this.totalFailures = totalFailures;
            this.averageProcessingTime = averageProcessingTime;
            this.maxProcessingTime = maxProcessingTime;
            this.minProcessingTime = minProcessingTime;
            this.throughput = throughput;
            this.averageSuccessRate = averageSuccessRate;
        }
        
        @Override
        public long getTotalBatches() { return totalBatches; }
        
        @Override
        public long getTotalProcessed() { return totalProcessed; }
        
        @Override
        public long getTotalSuccess() { return totalSuccess; }
        
        @Override
        public long getTotalFailures() { return totalFailures; }
        
        @Override
        public double getAverageProcessingTime() { return averageProcessingTime; }
        
        @Override
        public long getMaxProcessingTime() { return maxProcessingTime; }
        
        @Override
        public long getMinProcessingTime() { return minProcessingTime; }
        
        @Override
        public double getThroughput() { return throughput; }
        
        @Override
        public double getAverageSuccessRate() { return averageSuccessRate; }
        
        @Override
        public String toString() {
            return String.format(
                "StreamBatchStatistics:\n" +
                "  Total Batches: %d\n" +
                "  Total Processed: %d\n" +
                "  Success: %d, Failures: %d\n" +
                "  Success Rate: %.2f%%\n" +
                "  Avg Processing Time: %.2fms\n" +
                "  Max/Min Time: %dms/%dms\n" +
                "  Throughput: %.2f ops/sec",
                totalBatches, totalProcessed, totalSuccess, totalFailures,
                averageSuccessRate * 100, averageProcessingTime,
                maxProcessingTime, minProcessingTime, throughput
            );
        }
    }
}
